import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;


public class LineIndexer {

	
	public static class LineIndexerMapper extends MapReduceBase implements Mapper {
	    private final static Text word = new Text();
	    private final static Text summary = new Text();
	    public void map(WritableComparable key, Writable val, OutputCollector output, Reporter reporter)
	        throws IOException {
	        String line = val.toString();
	        summary.set(key.toString() + ":" + line);
	        StringTokenizer itr = new StringTokenizer(line.toLowerCase());
	        while(itr.hasMoreTokens()) {
	            word.set(itr.nextToken());
	            output.collect(word, summary);
	        }
	    }
	}
	
	public static class LineIndexerReducer extends MapReduceBase implements Reducer {
	    @SuppressWarnings("unchecked")
		public void reduce(WritableComparable key, Iterator values, OutputCollector output, Reporter reporter) throws IOException {
	        boolean first = true;
	        StringBuilder toReturn = new StringBuilder();
	        while(values.hasNext()){
	            if(!first)
	                toReturn.append('^');
	            first=false;
	            toReturn.append(values.next().toString());
	        }
	        output.collect(key, new Text(toReturn.toString()));
	    }
	}
	
	public static void main(String[] args) throws IOException {
	    JobConf conf = new JobConf(LineIndexer.class);
	    conf.setJobName("LineIndexer");
	    // The keys are words (strings):
	    conf.setOutputKeyClass(Text.class);
	    // The values are offsets+line (strings):
	    conf.setOutputValueClass(Text.class);
	    conf.setMapperClass(LineIndexer.LineIndexerMapper.class);
	    conf.setReducerClass(LineIndexer.LineIndexerReducer.class);
	    conf.setInputFormat(DocIDTextInputFormat.class);
	    if (args.length < 2) {
	        System.out.println("Usage: LineIndexer <input path> <output path>");
	        System.exit(0);
	    }
	    conf.setInputPath(new Path(args[0]));
	    conf.setOutputPath(new Path(args[1]));
	    JobClient.runJob(conf);
	}

}
